package cn.getech.data.development.task

import cn.getech.data.development.bean.FlinkStreamSyncHiveObj
import cn.getech.data.development.enums.{CustomJDBCType, OutputFormat}
import cn.getech.data.development.sink.sync.{FlinkRealtimeCollectHive, FlinkRealtimeCollectKudu, FlinkRealtimeCollectTrait}
import cn.getech.data.development.utils.FlinkUtils
import org.apache.commons.lang3.SystemUtils
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.types.Row
import org.slf4j.{Logger, LoggerFactory}

/**
 * hive sync
  * 实时采集主模块
 */
object FlinkStreamSyncHiveMain {
  private val logger: Logger = LoggerFactory.getLogger(this.getClass)
  private var obj: FlinkStreamSyncHiveObj = null

  def main(args: Array[String]): Unit = {
    // 读取资源
    setupDataSource(args)

    val jdbcType = CustomJDBCType.withName(obj.jdbc.typename)

    val env = FlinkUtils.env(obj.jobName)
    val tEnv: StreamTableEnvironment = FlinkUtils.createStreamTableEnv(env)

    var collectCls: FlinkRealtimeCollectTrait[Row] = null
    jdbcType match {
      case CustomJDBCType.hive =>
        logger.info("Loading hive operation.........")
        collectCls = new FlinkRealtimeCollectHive(obj)
        collectCls.overwriteTable
      case _ =>
        logger.info("Loading kudu operation.........")
        collectCls = new FlinkRealtimeCollectKudu(obj)
    }


    if (obj.kafkaResource.formatType == OutputFormat.csv || obj.kafkaResource.formatType == OutputFormat.text) {
      logger.info("Loading csv/text sink.........")
      collectCls.sinkCsv(tEnv)
    } else {
      logger.info("Loading json sink.........")
      collectCls.sinkJson(env)
    }

    env.execute(obj.jobName)
  }

  // 处理资源
  private def setupDataSource(args: Array[String]): Unit = {
    if (SystemUtils.IS_OS_WINDOWS)
      obj = FlinkStreamSyncHiveObj("data-development-streaming-flink-job/src/main/resources/flink-streaming-kudu-sync.json")
    else {
      logger.info("json file path: ", args(0))
      obj = FlinkStreamSyncHiveObj(args(0))
    }
    obj.jsonParse
  }


}
